package com.iflytek.controller;

import com.iflytek.bean.MonitorInfo;
import com.iflytek.schema.JSONDeserializationSchema;
import com.iflytek.utils.HBaseUtils;
import com.iflytek.utils.JdbcUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;


public class _05IllegalVehiclesTrackV2 {

    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);  //设置并行度

        //2.设置数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","hadoop102:9092");
        properties.setProperty("group.id","g4");


        FlinkKafkaConsumer<MonitorInfo> consumer = new FlinkKafkaConsumer<MonitorInfo>("topic-car",   new JSONDeserializationSchema<>(MonitorInfo.class),properties);
        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);

        SingleOutputStreamOperator<MonitorInfo> resultDs = ds1.filter(new RichFilterFunction<MonitorInfo>() {

            ResultSet resultSet = null;
            PreparedStatement preparedStatement = null;
            Connection connection = null;


            @Override
            public void open(Configuration parameters) throws Exception {
                connection = JdbcUtils.getconnection();
                preparedStatement = connection.prepareStatement("select * from t_violation_list where car = ?");

            }

            @Override
            public void close() throws Exception {
                JdbcUtils.release(resultSet, preparedStatement, connection);
            }

            @Override
            public boolean filter(MonitorInfo value) throws Exception {
                preparedStatement.setString(1, value.getCar());
                resultSet = preparedStatement.executeQuery();
                return resultSet.next();
            }
        });

        resultDs.print();
        // 将数据放入hbase中
        resultDs.addSink(new HBaseSink("t_track_info"));
        // 可以通过遍历resultDs，然后通过hbase原生API将数据保存到hbase中。

        env.execute();
    }

    public static class HBaseSink extends RichSinkFunction<MonitorInfo> {

        private final String tableName;
        private transient Table table;
        private transient Admin admin;

        public HBaseSink(String tableName) {
            this.tableName = tableName;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            // 获取表引用，如果表不存在则创建
            admin = HBaseUtils.connection.getAdmin();
            TableName hTableName = TableName.valueOf(tableName);

            if (!admin.tableExists(hTableName)) {
                // 表描述符
                TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(hTableName);
                // 列族描述符
                ColumnFamilyDescriptorBuilder cfBuilder1 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("cf1"));
                // 添加列族
                tableBuilder.setColumnFamily(cfBuilder1.build());

                admin.createTable(tableBuilder.build());
                System.out.println(hTableName.getNameAsString() + "表创建成功！");
            }

            table = HBaseUtils.getTable(hTableName.toString());
        }

        @Override
        public void invoke(MonitorInfo value, Context context) throws Exception {
            // 创建Put对象
            Put put = new Put(Bytes.toBytes(value.getCar()+"_"+value.getActionTime()));
            // 添加列
            System.out.println(value.toString());
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("actionTime"), Bytes.toBytes(value.getActionTime()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("areaId"), Bytes.toBytes(value.getAreaId()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("cameraId"), Bytes.toBytes(value.getCameraId()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("car"), Bytes.toBytes(value.getCar()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("monitorId"), Bytes.toBytes(value.getMonitorId()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("roadId"), Bytes.toBytes(value.getRoadId()));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("speed"), Bytes.toBytes(value.getSpeed()));
            // 写入数据
            table.put(put);
        }

        @Override
        public void close() throws Exception {
            super.close();
            HBaseUtils.closeTable(table);
            HBaseUtils.close(admin);
            HBaseUtils.connection.close();
        }
    }
}
